package com.study.flink.scala.day01_wc

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala._
import org.apache.flink.api.common.typeinfo.TypeInformation

object StreamWordCount {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //source
    val lines: DataStream[String] = env.socketTextStream("node02", 8888)

    //transform
    val words: DataStream[String] = lines.flatMap(_.split(" "))
    val wordAndOne: DataStream[(String, Int)] = words.map((_, 1))
    val keyed: KeyedStream[(String, Int), Tuple] = wordAndOne.keyBy(0)
    val summed: DataStream[(String, Int)] = keyed.sum(1)

    //sink
    summed.print()

    env.execute("StreamWordCount-scala")

  }

}
